use bytes::{Buf, BufMut};
use parking_lot::Mutex;
use std::fs::{File, OpenOptions};
use std::hash::Hasher;
use std::io;
use std::io::{BufReader, BufWriter, Read, Write};
use std::path::Path;
use std::sync::Arc;

pub struct Wal {
    path: String,
    file: Arc<Mutex<BufWriter<File>>>,
}

impl Wal {
    pub fn create(path: impl AsRef<Path>) -> io::Result<Self> {
        let file = Arc::new(Mutex::new(BufWriter::new(
            OpenOptions::new()
                .read(true)
                .create_new(true)
                .write(true)
                .open(&path)?,
        )));
        Ok(Self {
            path: path.as_ref().to_string_lossy().to_string(),
            file,
        })
    }

    pub fn write(&self, key: &[u8], value: &[u8]) -> io::Result<()> {
        let mut file = self.file.lock();
        let mut buf: Vec<u8> =
            Vec::with_capacity(key.len() + value.len() + std::mem::size_of::<u16>());
        let mut hasher = crc32fast::Hasher::new();
        hasher.write_u16(key.len() as u16);
        buf.put_u16(key.len() as u16);
        hasher.write(key);
        buf.put_slice(key);
        hasher.write_u16(value.len() as u16);
        buf.put_u16(value.len() as u16);
        buf.put_slice(value);
        hasher.write(value);
        // add checksum: week 2 day 7
        buf.put_u32(hasher.finalize());
        file.write_all(&buf)?;
        Ok(())
    }

    pub fn iter(&self) -> io::Result<WalIterator> {
        WalIterator::new(&self.path)
    }

    pub fn sync(&self) -> io::Result<()> {
        let mut file = self.file.lock();
        file.flush()?;
        file.get_mut().sync_all()?;
        Ok(())
    }
}

pub struct WalIterator {
    reader: BufReader<std::fs::File>,
    buffer: Vec<u8>,
    position: usize,
    end: usize,
}

impl WalIterator {
    pub fn new(path: impl AsRef<Path>) -> io::Result<Self> {
        let file = OpenOptions::new().read(true).append(true).open(path)?;
        Ok(Self {
            reader: BufReader::new(file),
            buffer: vec![0u8; 4096],
            position: 0,
            end: 0,
        })
    }

    fn refill_buffer(&mut self) -> io::Result<()> {
        self.position = 0;
        self.end = self.reader.read(&mut self.buffer)?;
        if self.end == 0 {
            return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "EOF reached"));
        }
        Ok(())
    }

    fn next_entry(&mut self) -> io::Result<Option<(Vec<u8>, Vec<u8>)>> {
        if self.position >= self.end {
            self.refill_buffer().ok();
        }

        let mut rbuf: &[u8] = &self.buffer[self.position..self.end];
        if rbuf.remaining() < 6 {
            return Ok(None); // Not enough data to read key_len and value_len
        }

        let mut hasher = crc32fast::Hasher::new();
        let key_len = rbuf.get_u16() as usize;
        hasher.write_u16(key_len as u16);

        if rbuf.remaining() < key_len {
            return Ok(None); // Not enough data to read the key
        }
        let key = rbuf[..key_len].to_vec();
        hasher.write(&key);
        rbuf.advance(key_len);

        if rbuf.remaining() < 2 {
            return Ok(None); // Not enough data to read value_len
        }
        let value_len = rbuf.get_u16() as usize;
        hasher.write_u16(value_len as u16);

        if rbuf.remaining() < value_len {
            return Ok(None); // Not enough data to read the value
        }
        let value = rbuf[..value_len].to_vec();
        hasher.write(&value);
        rbuf.advance(value_len);

        if rbuf.remaining() < 4 {
            return Ok(None); // Not enough data to read checksum
        }
        let checksum = rbuf.get_u32();
        if hasher.finalize() != checksum {
            return Err(io::Error::new(io::ErrorKind::Other, "checksum mismatch"));
        }

        self.position = self.end - rbuf.remaining();
        Ok(Some((key, value)))
    }
}

impl Iterator for WalIterator {
    type Item = (Vec<u8>, Vec<u8>);

    fn next(&mut self) -> Option<Self::Item> {
        match self.next_entry() {
            Ok(Some(entry)) => Some(entry),
            Ok(None) => None,
            Err(_err) => None,
        }
    }
}

#[cfg(test)]
mod test {
    use crate::Wal;
    use tempfile::tempdir;

    #[test]
    fn test() {
        let dir = tempdir().unwrap();
        let file_path = dir.path().join("my-temporary-wal");
        let wal = Wal::create(&file_path).unwrap();
        wal.write(b"a", b"a").unwrap();
        wal.write(b"b", b"b").unwrap();
        wal.write(b"c", b"c").unwrap();
        wal.write(b"d", b"d").unwrap();
        wal.sync().unwrap();
        for (k, v) in wal.iter().unwrap() {
            println!(
                "{} {}",
                String::from_utf8(k).unwrap(),
                String::from_utf8(v).unwrap()
            )
        }
        for (k, v) in wal.iter().unwrap() {
            println!(
                "{} {}",
                String::from_utf8(k).unwrap(),
                String::from_utf8(v).unwrap()
            )
        }
    }
}
